WITH barscans AS (
      -- 揽收 taking
      SELECT billcode     AS waybill_no,         -- 运单号
             scansitecode AS network_code,       -- 网点编号
             scantime     AS scantime,           -- 扫描时间
             weight       AS weight,             -- 重量
             NULL         AS next_network_code,  --all null
             NULL         AS shipment_no,        --all null
             NULL         AS package_code,       --all null
             scanuser     AS scanuser,           --操作扫描员
             "快件揽收"   AS scantype,           -- 扫描类型
             'taking'     AS table_type           -- 表类型
        FROM jms_ods.tab_barscan_taking
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 入仓 warehousing
      SELECT billcode     AS waybill_no,
             scansitecode AS network_code,
             scantime     AS scantime,
             weight       AS weight,
             NULL         AS next_network_code,
             NULL         AS shipment_no,
             NULL         AS package_code,       --all null
             scanuser     AS scanuser,           --操作扫描员
             "入仓扫描"     AS scantype,
             'warehousing'     AS table_type
        FROM jms_ods.tab_barscan_warehousing
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 建包 bagging
      SELECT billcode     AS waybill_no,
             scansitecode AS network_code,
             scantime     AS scantime,
             weight       AS weight,
             NULL         AS next_network_code,   --all null
             NULL         AS shipment_no,
             packagecode  AS package_code,       --所属袋号
             NULL         AS scanuser,           --操作扫描员
             "建包扫描"     AS scantype,
             'bagging'    AS table_type
        FROM jms_ods.tab_barscan_bagging
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
    -- 拆包 unbagging
         SELECT billcode     AS waybill_no,
                scansitecode AS network_code,
                scantime     AS scantime,
                weight       AS weight,
                NULL         AS next_network_code,   --all null
                NULL         AS shipment_no,
                packagecode  AS package_code,       --所属袋号
                scanuser     AS scanuser,           --操作扫描员
                scantype     AS scantype,
                'unbagging'  AS table_type
           FROM jms_ods.tab_barscan_other
          WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
          and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
          and scantype = '拆包扫描'
       UNION ALL
      -- 卸车 unloading
      SELECT billcode                    AS waybill_no,
             scansitecode                AS network_code,
             scantime                    AS scantime,
             weight                      AS weight,
             NULL                        AS next_network_code,
             nullif(transfercode, '')    AS shipment_no,
             NULL                        AS package_code,       --all null
             NULL                        AS scanuser,           --操作扫描员
             "卸车扫描"                  AS scantype,
             'unloading'                 AS table_type
        FROM jms_ods.tab_barscan_unloading
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 到件 arrival
      SELECT billcode                    AS waybill_no,
             scansitecode                AS network_code,
             scantime                    AS scantime,
             weight                      AS weight,
             NULL                        AS next_network_code,
             nullif(transfercode, '')    AS shipment_no,
             NULL                        AS package_code,       --all null
             scanuser                    AS scanuser,           --操作扫描员
             scantype                    AS scantype,
             'arrival'                   AS table_type
        FROM jms_ods.tab_barscan_arrival
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 装车 loading
      SELECT billcode                    AS waybill_no,
             scansitecode                AS network_code,
             scantime                    AS scantime,
             weight                      AS weight,
             nullif(nextstationcode, '') AS next_network_code,
             nullif(transfercode, '')    AS shipment_no,
             NULL                        AS package_code,
             NULL                        AS scanuser,
             "装车扫描"                  AS scantype,
             'loading'                   AS table_type
        FROM jms_ods.tab_barscan_loading
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 发件 send
      SELECT billcode                    AS waybill_no,
             scansitecode                AS network_code,
             scantime                    AS scantime,
             weight                      AS weight,
             nullif(nextstationcode, '') AS next_network_code,
             nullif(transfercode, '')    AS shipment_no,
             NULL                        AS package_code,       --all null
             scanuser                    AS scanuser,           --操作扫描员
             "发件扫描"                  AS scantype,
             'send'                      AS table_type
        FROM jms_ods.tab_barscan_send
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 派件 deliver
      SELECT billcode     AS waybill_no,
             scansitecode AS network_code,
             scantime     AS scantime,
             weight       AS weight,
             NULL         AS next_network_code,
             NULL         AS shipment_no,
             NULL         AS package_code,       --all null
             scanuser     AS scanuser,           --操作扫描员
             "出仓扫描"   AS scantype,
             'deliver'    AS table_type
        FROM jms_ods.tab_barscan_deliver
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 签收 sign
      SELECT billcode     AS waybill_no,
             scansitecode AS network_code,
             delivertime  AS scantime,
             realweight   AS weight,
             NULL         AS next_network_code,
             NULL         AS shipment_no,
             NULL         AS package_code,
             deliveruser  AS scanuser,
             '签收扫描'   AS scantype,
             'sign'       AS table_type
        FROM jms_ods.tab_barscan_sign
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(delivertime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       UNION ALL
      -- 出入库 collect
      SELECT billcode        AS waybill_no,
             scansitecode    AS network_code,
             scantime        AS scantime,
             weight          AS weight,
             NULL            AS next_network_code,
             NULL            AS shipment_no,
             NULL            AS package_code,
             scanuser        AS scanuser,
             scantype        AS scantype,
             case when scantype='入库扫描' then 'collecting1'
                  when scantype='出库扫描' then 'collecting2'
                  else 'collecting3'  end AS table_type -- 入库扫描 代理点收入扫描 快件取出扫描 出库扫描
        FROM jms_ods.tab_barscan_collect
       WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
       and to_date(scantime) BETWEEN date_add('{{ execution_date | cst_ds }}', -5) AND '{{ execution_date | cst_ds }}'
         AND scantype != '快件取出扫描' and scansitecode is not null ),
       shipment     AS (SELECT shipment_no                    AS shipment_no,             -- 转运任务单号
                                network_code                  AS network_code,            -- 网点编码
                                min(planned_departure_time)   AS planned_departure_time,  -- 计划发车时间
                                min(planned_arrival_time)     AS planned_arrival_time,    -- 计划到达时间
                                min(actual_departure_time)    AS actual_departure_time,   -- 实际发车时间
                                min(actual_arrival_time)      AS actual_arrival_time      -- 实际到达时间
                           FROM (SELECT shipment_no, network_code, actual_departure_time, actual_arrival_time, planned_arrival_time, planned_departure_time
                                   FROM jms_ods.yl_tmsnew_branch_shipment_stop
                                  WHERE dt = '{{ execution_date | cst_ds }}'
                                    AND (actual_arrival_time >= date_add('{{ execution_date | cst_ds }}', -1) OR actual_departure_time >= date_add('{{ execution_date | cst_ds }}', -1)
                                     OR planned_arrival_time >= date_add('{{ execution_date | cst_ds }}', -1) OR planned_departure_time >= date_add('{{ execution_date | cst_ds }}', -1))
                                  UNION ALL
                                 SELECT shipment_no, network_code, actual_departure_time, actual_arrival_time, planned_arrival_time, planned_departure_time
                                   FROM jms_ods.yl_tmsnew_tms_shipment_stop
                                  WHERE dt = '{{ execution_date | cst_ds }}'
                                    AND (actual_arrival_time >= date_add('{{ execution_date | cst_ds }}', -1) OR actual_departure_time >= date_add('{{ execution_date | cst_ds }}', -1)
                                     OR planned_arrival_time >= date_add('{{ execution_date | cst_ds }}', -1) OR planned_departure_time >= date_add('{{ execution_date | cst_ds }}', -1))) AS stops
                          GROUP BY shipment_no, network_code),
       network     AS (select
                             case when type_id=335 and name like '%加盟商%'  then 1
                                  when type_id=335 and name like '%中心%'    then 2
                                  when type_id=336 and is_entrepot=1         then 3
                                  when type_id=336 and is_entrepot=2         then 4
                                  when type_id=334                           then 5
                                  when type_id=22                            then 6
                                  else 0  end as network_type,
                                  code
                       from jms_dim.dim_lmdm_sys_network),
       wide_barscan  AS (SELECT barscans.waybill_no                                           AS waybill_no,
                                barscans.network_code                                         AS network_code,
                                barscans.table_type                                            AS table_type,
                                barscans.scantime                                             AS scantime,
                                barscans.weight                                               AS weight,
                                barscans.next_network_code                                    AS next_network_code,
                                barscans.shipment_no                                          AS shipment_no,
                                barscans.package_code                                         AS package_code,
                                barscans.scanuser                                             AS scanuser,
                                barscans.scantype                                             AS scantype,
                                network.network_type                                          AS network_type,
                                next_network.network_type                                     AS next_network_type,
                                shipment.planned_departure_time                               AS planned_departure_time,  -- 计划发车时间
                                shipment.planned_arrival_time                                 AS planned_arrival_time,    -- 计划到达时间
                                shipment.actual_departure_time                                AS actual_departure_time,   -- 实际发车时间
                                shipment.actual_arrival_time                                  AS actual_arrival_time      -- 实际到达时间
                           FROM barscans
                           LEFT JOIN network AS network
                             ON barscans.network_code = network.code
                           LEFT JOIN network AS next_network
                             ON coalesce(barscans.next_network_code,barscans.waybill_no) = next_network.code
                           left join shipment
                               on coalesce(barscans.shipment_no,barscans.network_code) = shipment.shipment_no
                                  and barscans.network_code = shipment.network_code
                         ) ,
       collected  AS (SELECT waybill_no,
                                network_code,
                                table_type,
                                network_type,
                                named_struct('scantime', scantime,
                                            'weight', weight,
                                            'next_network_code', next_network_code,
                                            'shipment_no', shipment_no,
                                            'package_code', package_code,
                                            'scanuser', scanuser,
                                            'planned_departure_time', planned_departure_time,
                                            'planned_arrival_time', planned_arrival_time,
                                            'actual_departure_time', actual_departure_time,
                                            'actual_arrival_time', actual_arrival_time,
                                            'scantype', scantype,
                                            'next_network_type', next_network_type
                                           ) AS trace_wid
                           FROM wide_barscan),
        scantypedim AS (              --运单，网点，操作类型，操作次数，是否中心，是否集散， 最早、最晚的相关字段
                         SELECT waybill_no,
                                network_code,
                                table_type,
                                count(1) as cnt ,
                                max(network_type) as network_type,    --唯一，max 或者min都可，值为true 或者false
                                max(trace_wid) as max_wid,       --取运单在该网点下的 该操作中的最晚操作信息
                                min(trace_wid) as min_wid       --取运单在该网点下的 该操作中的最早操作信息
                        from collected
                        group by waybill_no,
                                network_code,
                                table_type
        ),
        networkdim AS (
                         SELECT waybill_no,
                                network_code,
                                max(network_type)                                                              as network_type, --唯一，max 或者min都可，值为true 或者false
                                sum(cnt)                                                                       as all_cnt ,   --该运单在该网点 所有的操作次数，通过这个可以维护运单黑名单
                                max(max_wid)                                                                   as network_later_cols, --取运单在该网点下的所有操作中的最晚操作信息
                                min(min_wid)                                                                   as network_early_cols, --取运单在该网点下的所有操作中的最早操作信息
                                max(if(table_type='taking',max_wid.scantime,null))                             as taking_later_scantime,  --取该运单在该网点 的taking操作中的最晚操作信息
                                min(if(table_type='taking',min_wid,null))                                      as taking_early_cols, --取该运单在该网点 的taking操作中的最早操作信息
                                max(if(table_type='warehousing',max_wid.scantime,null))                        as warehousing_later_scantime,
                                min(if(table_type='warehousing',min_wid,null))                                 as warehousing_early_cols,
                                min(if(table_type='bagging',min_wid.scantime,null))                            as bagging_early_scantime,
                                max(if(table_type='bagging',max_wid,null))                                     as bagging_later_cols,
                                --添加拆包操作 2021-09-17
                                min(if(table_type='unbagging',min_wid.scantime,null))                          as unbagging_early_scantime,
                                max(if(table_type='unbagging',max_wid,null))                                   as unbagging_later_cols,
                                max(if(table_type='arrival',max_wid.scantime,null))                            as arrival_later_scantime,
                                min(if(table_type='arrival',min_wid,null))                                     as arrival_early_cols,
                                max(if(table_type='unloading',max_wid.scantime,null))                          as unloading_later_scantime,
                                min(if(table_type='unloading',min_wid,null))                                   as unloading_early_cols,
                                min(if(table_type='loading',min_wid.scantime,null))                            as loading_early_scantime,
                                max(if(table_type='loading',max_wid,null))                                     as loading_later_cols,
                                min(if(table_type='send',min_wid.scantime,null))                               as send_early_scantime,
                                max(if(table_type='send',max_wid,null))                                        as send_later_cols,
                                min(if(table_type='deliver',min_wid.scantime,null))                            as deliver_early_scantime,
                                max(if(table_type='deliver',max_wid,null))                                     as deliver_later_cols,
                                max(if(table_type='sign',max_wid.scantime,null))                               as sign_later_scantime,
                                min(if(table_type='sign',min_wid,null))                                        as sign_early_cols,
                                min(if(table_type in ('sign','collecting2'),min_wid,null))                     as actual_sign_cols,
                                min(if(table_type in ('sign','collecting1','collecting2','collecting3'),min_wid,null))  as aging_sign_cols,
                                min(if(table_type in ('taking','warehousing'),min_wid,null))                   as actual_taking_cols,
                                to_date(max(max_wid).scantime)                                                 as dt
                        from scantypedim
                               group by waybill_no,
                                network_code
                                 having to_date(max(max_wid).scantime) = '{{ execution_date | cst_ds }}'
        )
            insert overwrite table jms_dwd.dwd_s01_whole_operations_basic_mid_dt
         SELECT                 waybill_no                                                                    as waybill_no,
                                networkdim.network_code                                                       as network_code,
                                network_type                                                                  as network_type,
                                all_cnt                                                                       as all_cnt ,
                                network_later_cols.scantime                                                   as network_later_scantime,
                                network_early_cols.scantime                                                   as network_early_scantime,
                                network_later_cols.scantype                                                   as network_later_scantype,
                                network_early_cols.scantype                                                   as network_early_scantype,
                                actual_taking_cols.scantime                                                   as actual_taking_scantime,
                                actual_taking_cols.weight                                                     as actual_taking_weight,
                                actual_taking_cols.scantype                                                   as actual_taking_scantype,
                                actual_taking_cols.scanuser                                                   as actual_taking_scanuser,
                                taking_later_scantime                                                         as taking_later_scantime,
                                taking_early_cols.scantime                                                    as taking_early_scantime,
                                taking_early_cols.weight                                                      as taking_early_weight,
                                taking_early_cols.scanuser                                                    as taking_early_scanuser,
                                warehousing_later_scantime                                                    as warehousing_later_scantime,
                                warehousing_early_cols.scantime                                               as warehousing_early_scantime,
                                warehousing_early_cols.weight                                                 as warehousing_early_weight,
                                warehousing_early_cols.scanuser                                               as warehousing_early_scanuser,
                                bagging_early_scantime                                                        as bagging_early_scantime,
                                bagging_later_cols.scantime                                                   as bagging_later_scantime,
                                bagging_later_cols.package_code                                               as bagging_later_package_code,
                                unloading_later_scantime                                                      as unloading_later_scantime,
                                unloading_early_cols.scantime                                                 as unloading_early_scantime,
                                unloading_early_cols.weight                                                   as unloading_early_weight,
                                arrival_later_scantime                                                        as arrival_later_scantime,
                                arrival_early_cols.scantime                                                   as arrival_early_scantime,
                                arrival_early_cols.weight                                                     as arrival_early_weight,
                                arrival_early_cols.scanuser                                                   as arrival_early_scanuser,

                                coalesce(unloading_early_cols.shipment_no            ,arrival_early_cols.shipment_no           )                               as unloading_arrival_shipment_no,
                                coalesce(unloading_early_cols.planned_departure_time ,arrival_early_cols.planned_departure_time)                               as unloading_arrival_planned_departure_time,
                                coalesce(unloading_early_cols.planned_arrival_time   ,arrival_early_cols.planned_arrival_time  )                               as unloading_arrival_planned_arrival_time,
                                coalesce(unloading_early_cols.actual_departure_time  ,arrival_early_cols.actual_departure_time )                               as unloading_arrival_actual_departure_time,
                                coalesce(unloading_early_cols.actual_arrival_time    ,arrival_early_cols.actual_arrival_time   )                               as unloading_arrival_actual_arrival_time,
                                coalesce(unloading_early_cols.next_network_code      ,arrival_early_cols.next_network_code     )                               as unloading_arrival_previous_network_code,
                                coalesce(unloading_early_cols.next_network_type      ,arrival_early_cols.next_network_type     )                               as unloading_arrival_previous_network_type,

                                loading_early_scantime                                                        as loading_early_scantime,
                                loading_later_cols.scantime                                                   as loading_later_scantime,
                                loading_later_cols.weight                                                     as loading_later_weight,
                                send_early_scantime                                                           as send_early_scantime,
                                send_later_cols.scantime                                                      as send_later_scantime,
                                send_later_cols.weight                                                        as send_later_weight,
                                send_later_cols.scanuser                                                      as send_later_scanuser,

                                coalesce(loading_later_cols.shipment_no              ,send_later_cols.shipment_no             )                                 as loading_send_shipment_no,
                                coalesce(loading_later_cols.planned_departure_time   ,send_later_cols.planned_departure_time  )                                 as loading_send_planned_departure_time,
                                coalesce(loading_later_cols.planned_arrival_time     ,send_later_cols.planned_arrival_time    )                                 as loading_send_planned_arrival_time,
                                coalesce(loading_later_cols.actual_departure_time    ,send_later_cols.actual_departure_time   )                                 as loading_send_actual_departure_time,
                                coalesce(loading_later_cols.actual_arrival_time      ,send_later_cols.actual_arrival_time     )                                 as loading_send_actual_arrival_time,
                                coalesce(loading_later_cols.next_network_code        ,send_later_cols.next_network_code       )                                 as loading_send_next_network_code,
                                coalesce(loading_later_cols.next_network_type        ,send_later_cols.next_network_type       )                                 as loading_send_next_network_type,

                                deliver_early_scantime                                                        as deliver_early_scantime,
                                deliver_later_cols.scantime                                                   as deliver_later_scantime,
                                deliver_later_cols.weight                                                     as deliver_later_weight,
                                deliver_later_cols.scanuser                                                   as deliver_later_scanuser,
                                sign_later_scantime                                                           as sign_later_scantime,
                                sign_early_cols.scantime                                                      as sign_early_scantime,
                                sign_early_cols.weight                                                        as sign_early_weight,
                                sign_early_cols.scanuser                                                      as sign_early_scanuser,
                                actual_sign_cols.scantime                                                     as actual_sign_scantime,
                                actual_sign_cols.weight                                                       as actual_sign_weight,
                                actual_sign_cols.scantype                                                     as actual_sign_scantype,
                                actual_sign_cols.scanuser                                                     as actual_sign_scanuser,
                                aging_sign_cols.scantime                                                      as aging_sign_scantime,
                                aging_sign_cols.weight                                                        as aging_sign_weight,
                                aging_sign_cols.scantype                                                      as aging_sign_scantype,
                                aging_sign_cols.scanuser                                                      as aging_sign_scanuser,
                                unbagging_early_scantime                                                      as unbagging_early_scantime,
                                unbagging_later_cols.scantime                                                 as unbagging_later_scantime,
                                unbagging_later_cols.package_code                                             as unbagging_later_package_code,
                                dt as dt
                        from networkdim
                        distribute by abs(hash(waybill_no)) % 180 ;